package SparkExercises;


import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import scala.Serializable;

import java.util.List;

public class zuoye {
    /**
     * 已知有一份新能源车辆行程数据trip.csv（见附件），
     * 它的第一列表示车辆的车架号（每辆车都是唯一的），
     * 第二列表示车型，第三列表示行程日期，第四列表示当天行驶的总里程（单位是公里），
     * 第五列表示当天行驶的总时长（单位是分钟）。
     * 请写一个spark程序（也就是使用到spark相关算子的都算spark程序），
     * 对这些数据按照总里程进行升序排序，对于总里程相同的，
     * 则按照总时长进行升序排序，并将排序之后的结果输出到控制台，如下图所示。
     * @param args
     */
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("zuoye").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD<String> rdd1 = sc.textFile("./data/trip.csv");

        JavaRDD<che> rdd2 = rdd1.map(new Function<String, che>() {
            @Override
            public che call(String s) throws Exception {
                String[] split = s.split(",");
                che che = new che(split[0], split[1], split[2], Integer.parseInt(split[3]), Integer.parseInt(split[4]));
                return che;
            }
        });


        Function<che, che> function = new Function<che, che>() {

            @Override
            public che call(che c) throws Exception {
                return c;
            }
        };
        JavaRDD<che> rdd3 = rdd2.sortBy(function, true, rdd1.getNumPartitions());

//        System.out.println(rdd2.collect());
        List<che> res = rdd3.collect();
        for (che i : res) {
            System.out.println(i);
        }

    }

}

class che implements Comparable<che>, Serializable {
    private String ch;
    private String cx;
    private String rq;
    private int zc;
    private int sc;

    public che(String ch, String cx, String rq, int zc, int sc) {
        this.ch = ch;
        this.cx = cx;
        this.rq = rq;
        this.zc = zc;
        this.sc = sc;
    }

    public che() {
    }

    public void setCh(String ch) {
        this.ch = ch;
    }

    public void setCx(String cx) {
        this.cx = cx;
    }

    public void setRq(String rq) {
        this.rq = rq;
    }

    public void setZc(int zc) {
        this.zc = zc;
    }

    public void setSc(int sc) {
        this.sc = sc;
    }

    public String getCh() {
        return ch;
    }

    public String getCx() {
        return cx;
    }

    public String getRq() {
        return rq;
    }

    public int getZc() {
        return zc;
    }

    public int getSc() {
        return sc;
    }

    @Override
    public String toString() {
        return "che{" +
                "ch='" + ch + '\'' +
                ", cx='" + cx + '\'' +
                ", rq='" + rq + '\'' +
                ", zc=" + zc +
                ", sc=" + sc +
                '}';
    }

    @Override
    public int compareTo(che o) {
        if (this.getZc() > o.getZc()) {
            return 1;
        } else if (this.getZc() == o.getZc()) {
            if (this.getSc() > o.getSc()) {
                return 1;
            } else if (this.getSc() == o.getSc()) {
                return 0;
            } else {
                return -1;
            }
        } else {
            return -1;
        }
    }

}